{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "06a6f7f7-932f-4bf0-b0e6-5dea77f01608",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Looking in indexes: https://pypi.tuna.tsinghua.edu.cn/simple\n",
      "Collecting confluent-kafka\n",
      "  Downloading https://pypi.tuna.tsinghua.edu.cn/packages/42/08/05be324ca1765e3bf46b4c5b2f1707941547ab3a0a4f8fef376b66344c90/confluent_kafka-2.3.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (4.0 MB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m4.0/4.0 MB\u001b[0m \u001b[31m13.5 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0ma \u001b[36m0:00:01\u001b[0m\n",
      "\u001b[?25hInstalling collected packages: confluent-kafka\n",
      "Successfully installed confluent-kafka-2.3.0\n"
     ]
    }
   ],
   "source": [
    "#!pip install confluent-kafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5ba91d7b-6d95-427a-a22d-181ade5198ad",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# %%\n",
    "import socket\n",
    "import threading\n",
    "import struct\n",
    "import json\n",
    "from confluent_kafka import Producer\n",
    "from datetime import datetime\n",
    "from IPython.display import clear_output,display"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e086fe8b-55af-4858-b38b-551b82891bf5",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "## 数据表头\n",
    "column_name_file = './g120_column_name_240313.txt'\n",
    "kafka_config = {'bootstrap.servers':'da-sever-b:9092,da-node1-b:9092,da-node2-b:9092'}\n",
    "kafka_topic = 'g120v1'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "19d144c7-91f7-4093-929e-2bd0d40f3794",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def column_name_mapping(column_name_file):\n",
    "    # 打开并读取文件\n",
    "    with open(column_name_file, 'r', encoding='utf-8') as f:\n",
    "        data = f.read()\n",
    "    # 将字符串按照换行符分割成列表\n",
    "    data_names = data.split('\\n')\n",
    "    # processed_names = [name.replace(\" \", \"_\").replace(\"%\", \"percent\").replace(\"℃\", \"degC\").replace(\".\", \"\").replace(\"Ω\",\"Ohm\").replace(\n",
    "    #     \"μ\", \"Micro_\").replace(\"/\", \"_per_\") for name in data_names]\n",
    "    # 创建一个字典来存储原始变量名和dummy变量名之间的映射关系\n",
    "    name_mapping = {original: f'a_{i}' for i, original in enumerate(data_names)}\n",
    "    # 使用dummy变量名来代替原始变量名\n",
    "    dummy_names = [name_mapping[name] for name in data_names]\n",
    "    return dummy_names, name_mapping\n",
    "\n",
    "def get_original_name(dummy_names, name_mapping):\n",
    "    # Reverse the name_mapping dictionary\n",
    "    reverse_mapping = {v: k for k, v in name_mapping.items()}\n",
    "    # Get the original name\n",
    "    original_name = reverse_mapping.get(dummy_names)\n",
    "    return original_name\n",
    "\n",
    "dummy_names, name_mapping=column_name_mapping(column_name_file)\n",
    "original_names = [get_original_name(name,name_mapping) for name in dummy_names]\n",
    "\n",
    "\n",
    "## 二进制向十进制转码，双精度浮点数\n",
    "def decode_data(data):\n",
    "    # 'f'表示浮点数，'>'表示大端字节序\n",
    "    format_str = '>' + 'd' * (len(data) // 8)  # 双精度浮点数解码\n",
    "    # format_str = '>' + 'f' * (len(data) // 4) # 单精度浮点数解码\n",
    "    decoded_data = struct.unpack(format_str, data)\n",
    "    return decoded_data\n",
    "\n",
    "\n",
    "\n",
    "## 多线程处理服务器端\n",
    "def delivery_report(err, msg):\n",
    "    if err is not None:\n",
    "        print(f\"Message delivery failed: {err}\")\n",
    "    #else:\n",
    "        #print(f\"Message delivered to {msg.topic()} [{msg.partition()}]\")\n",
    "\n",
    "def handle_client(conn, addr, dummy_names,kafka_config=kafka_config,kafla_topic=kafka_topic):\n",
    "    print('Connected by', addr)\n",
    "\n",
    "    # 创建一个Kafka的生产者\n",
    "\n",
    "    p = Producer(kafka_config)\n",
    "\n",
    "    while True:\n",
    "        data = conn.recv(8192 * 10)\n",
    "        if not data:\n",
    "            break\n",
    "        #print(\"Received:\", data)\n",
    "        conn.sendall(data)  # 将接受到的数据反馈回客户端，视情况关闭此功能\n",
    "        decoded_data = decode_data(data)\n",
    "        #print(\"decode_data:\", decoded_data)\n",
    "\n",
    "        # 创建一个字典，其中键是数据的名称，值是解码后的数据\n",
    "        data_dict = dict(zip(dummy_names, decoded_data))\n",
    "\n",
    "        # 将第一列的浮点数数据（假设是时间戳）转换为字符串\n",
    "        time_float = decoded_data[0]\n",
    "        time_str = str(time_float)\n",
    "\n",
    "        # 将字符串转换为日期时间类型\n",
    "        time_obj = datetime.strptime(time_str, '%Y%m%d%H%M%S.%f')\n",
    "\n",
    "        timestamp_ms=int(time_obj.timestamp()*1000)\n",
    "        #print(\"timestamp_ms=\",timestamp_ms)\n",
    "        # 将日期时间类型转换为字符串，精确到毫秒\n",
    "        time_str = time_obj.strftime('%Y-%m-%d %H:%M:%S.%f')\n",
    "        #print(\"time_str=\", time_str)\n",
    "\n",
    "        # 将时间字符串添加到数据字典中，键是dummy_names[0]\n",
    "        data_dict[dummy_names[0]] = timestamp_ms\n",
    "\n",
    "        # 将字典转换为JSON格式的字符串\n",
    "        json_data = json.dumps(data_dict)\n",
    "\n",
    "\n",
    "        # 输出JSON格式的数据\n",
    "        #print(\"JSON data:\", json_data)\n",
    "        clear_output(wait=True)\n",
    "        display(time_str)\n",
    "        display(json_data)\n",
    "\n",
    "        # 将JSON数据发送到Kafka\n",
    "        p.produce(kafla_topic, json_data, callback=delivery_report)\n",
    "        p.flush()\n",
    "        # p.poll(0)\n",
    "\n",
    "    conn.close()\n",
    "\n",
    "def start_server(server_host='0.0.0.0', server_port=30001, dummy_names=None,kafla_config=kafka_config,kafka_topic=kafka_topic):\n",
    "    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:\n",
    "        s.bind((server_host, server_port))\n",
    "        s.listen()\n",
    "        print(f\"Server started at {server_host}:{server_port}. Waiting for connection...\")\n",
    "        while True:\n",
    "            conn, addr = s.accept()\n",
    "            client_thread = threading.Thread(target=handle_client, args=(conn, addr, dummy_names,kafla_config,kafka_topic))\n",
    "            client_thread.start()\n",
    "            #clear_output(wait=True)  \n",
    "\n",
    "#%%\n",
    "##数据发送客户端\n",
    "# import socket\n",
    "# #\n",
    "# def start_client(server_host='192.168.10.12', server_port=30001):\n",
    "#     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:\n",
    "#         s.connect((server_host, server_port))\n",
    "#         while True:\n",
    "#             # 发送数据\n",
    "#             s.sendall(b'Hello, server')\n",
    "#             # 接收数据\n",
    "#             data = s.recv(1024)\n",
    "#             print(\"Received:\", data)\n",
    "#\n",
    "# if __name__ == \"__main__\":\n",
    "#     start_client()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7ce349e6-cc1d-4251-8bb5-f92467aadf7a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "name_mapping"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e83bb96c-ec25-4de9-a550-13a5be36b252",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "original_names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d1044b25-19f4-421f-ada0-d83ffd34b3bb",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "if __name__ == \"__main__\":\n",
    "    start_server(dummy_names=dummy_names,kafla_config=kafka_config,kafka_topic=kafka_topic)    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ad80bf6e-78e8-4388-97c9-402ecc317947",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "py38-env",
   "language": "python",
   "name": "py38"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
